package com.yugao.fintech.draper.message.queue.support;

import cn.hutool.core.thread.ThreadUtil;
import com.hazelcast.collection.IQueue;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.topic.ITopic;
import com.yugao.fintech.draper.message.queue.client.HazelcastMQTemplate;
import com.yugao.fintech.draper.message.queue.support.broadcast.AbstractHazelcastBroadcastMessageListener;
import com.yugao.fintech.draper.message.queue.support.broadcast.HazelcastBroadcastMessage;
import com.yugao.fintech.draper.message.queue.support.group.AbstractHazelcastGroupMessageListener;
import com.yugao.fintech.draper.message.queue.support.group.HazelcastGroupMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;

@Slf4j
@RequiredArgsConstructor
public class HazelcastMessageListenerProvider {

    private final HazelcastInstance hazelcastInstance;

    private final HazelcastMQTemplate hazelcastMQTemplate;

    private ExecutorService executorService = ThreadUtil.newExecutor(15);

    public <M extends HazelcastBroadcastMessage> void register(AbstractHazelcastBroadcastMessageListener listener) {
        listener.setMqTemplate(hazelcastMQTemplate);
        String topicName = listener.getTopic();
        ITopic<M> topic = hazelcastInstance.getTopic(topicName);
        topic.addMessageListener(listener);
    }

    public <M extends HazelcastGroupMessage> void register(AbstractHazelcastGroupMessageListener listener) {
        listener.setMqTemplate(hazelcastMQTemplate);
        String groupName = listener.getGroup();
        IQueue<M> queue = hazelcastInstance.getQueue(groupName);
        Thread worker = new Thread(() -> {
            while (true) {
                try {
                    M m = queue.take();
                    executorService.execute(() -> {
                        listener.process(m);
                    });
                } catch (Throwable e) {
                    log.info("{} 线程异常: {}", Thread.currentThread().getName(), e.getLocalizedMessage());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException interruptedException) {
                        log.info("{} 线程异常: {}", Thread.currentThread().getName(), interruptedException.getLocalizedMessage());
                    }
                }
            }
        }, "HazelcastConsumerThread-" + groupName);

        worker.start();
    }
}
